/* 
 * Quechua - the lightweight data mining framework
 *
 * Copyright (C) 2012 Marek Denis <quechua@octogan.net>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "../include/threads.h"
#include "../include/quechua.h"

Threads::Threads(int threads_num) {
    threads_num = (threads_num<2?2:threads_num); //safe-check
    pool = new g_threadpool(threads_num);
    LOG(INFO) << "Quechua will be using " << threads_num << " threads";

};

Threads::Threads() {
    int threads_num = calculate_threads_number();
    pool = new g_threadpool(threads_num);
    LOG(INFO) << "Quechua will be using " << threads_num << " threads";
};

void Threads::prepare() {
    cmd_watcher.set<Threads,&Threads::handle_command_cb>(this);
    cmd_watcher.set(*g_quechua->loop);
};

void Threads::start() {
    cmd_watcher.start();
}

void Threads::stop() {
    cmd_watcher.stop();                                                                    
    pool->shutdown(true);
};

Threads::~Threads() {
    delete pool;
};

void Thread::run() {
#ifdef DEBUG_MSG
    LOG(DEBUG) << "Running Thread at: " << this;
#endif
    datapack_ptr thread_result  = command->function(command->data,
                                                    command->object,
                                                    command->step);
    // prepare response command
    command_ptr response(new command_s);
    response->flow = command->flow;
    response->step = command->step;
    response->function = NULL;
    response->object = NULL;
    
    if(!thread_result) {
        LOG(ERROR) << "Error, nothing returned from computing module,"
                      " setting step to DONE, discarding all data";
        response->step = DONE;
    } 
    else {
        response->data = thread_result;
    } 
    finish(response);
    return;
};

Thread::~Thread() {
#ifdef DEBUG_MSG
    LOG(DEBUG) << "Deleting Thread at: " << this;
#endif
};

void Thread::finish(command_ptr cmd) {
    g_quechua->threads->handle_command(cmd);
    delete this;
};

void Threads::RUN_IN_THREAD(datapack_ptr data, 
                            void* object, 
                            Workflow* flow, thread_func_ptr function) {
#ifdef DEBUG_MSG
    LOG(DEBUG) << "RUN_IN_THREAD";
#endif
    command_ptr cmd(new command_s);
    cmd->data = data;
    cmd->function = function;
    cmd->step = flow->current_step();
    cmd->object = object;
    cmd->flow  = flow;
    
    data.reset();

    if(pool==NULL) {
        LOG(CRITIC) << "g_quechua->threads->pool is NULL";
        return;
    }

    pool->push(sigc::mem_fun(new Thread(cmd),&Thread::run));
    return;
};

/**
 * Called by a separated thread
 **/ 
void Threads::handle_command(command_ptr cmd) {
    mutex.lock(); 
#ifdef DEBUG_MSG
    {LOG(DEBUG) << "handle_command() mutex.lock()";}
#endif
        cmds.push(cmd);
#ifdef DEBUG_MSG
    {LOG(WARN) << "handle_command() mutex.unlock()";}
#endif
    mutex.unlock();
#ifdef DEBUG_MSG
    {LOG(DEBUG) << "cmd_watcher.send()";}
#endif
    cmd_watcher.send(); // notify libev about new commands
};

void Threads::handle_command_cb(ev::async &watcher, int revents) {
#ifdef DEBUG_MSG    
    {LOG(DEBUG) << "threads_s::handle_command_cb";};
#endif
    mutex.lock();
#ifdef DEBUG_MSG
    {LOG(WARN) << "handle_command_cb() mutex.lock()";}
#endif
   
     while(!cmds.empty()) {
      command_ptr cmd = cmds.front();
      cmds.pop();
      cmd->flow->notify(cmd->data,cmd->step);
//      cmd->data.reset();
      cmd.reset();
     }
#ifdef DEBUG_MSG
    {LOG(WARN) << "handle_command_cb() mutex.unlock()";}
#endif
    mutex.unlock();
};

int Threads::calculate_threads_number() {
   int threads = sysconf(_SC_NPROCESSORS_ONLN);
   threads = (threads <= 1?5:threads);
//   threads = 5;
   return threads;
};

